package com.dtp.core.thread;

import com.dtp.core.reject.RejectHandlerGetter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static com.dtp.common.em.QueueTypeEnum.buildLbq;

/**
 * OrderedDtpExecutor可以确保按照一定的顺序执行。适用于通过并行处理提高吞吐量，任务按一定顺序运行的场景。
 *
 * @author dragon-zhang
 *
 * @remark muse
 */
@Slf4j
public class OrderedDtpExecutor extends DtpExecutor {
    protected final AtomicInteger count = new AtomicInteger(0); // 计数器，用于choose方法中选择某一个Dtp执行器
    protected final List<DtpExecutor> executors = new ArrayList<>(); // Dtp执行器列表

    /** -------------------------------------构造函数------------------------------------- */
    public OrderedDtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), new AbortPolicy());
    }
    public OrderedDtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                              BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new AbortPolicy());
    }
    public OrderedDtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                              BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler);
    }
    public OrderedDtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                              BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        // 创建corePoolSize个线程池
        for (int i = 0; i < corePoolSize; i++) {
            executors.add(new DtpExecutor(1, 1, keepAliveTime, unit, buildLbq(getQueueName(), getQueueCapacity()),
                    buildThreadFactory(i), RejectHandlerGetter.getProxy(handler)));
        }
    }

    /** 执行任务（无返回值）*/
    @Override
    public void execute(Runnable command) {execute(null, command);}
    public void execute(Object arg, Runnable command) {choose(arg).execute(command);}

    /** 执行任务（有返回值Future）*/
    @Override
    public Future<?> submit(Runnable task) {return submit((Object) null, task);}
    @Override
    public <T> Future<T> submit(Runnable task, T result) {return submit(null, task, result);}
    @Override
    public <T> Future<T> submit(Callable<T> task) {return submit((Object) null, task);}
    public Future<?> submit(Object arg, Runnable task) {return choose(arg).submit(task);}
    public <T> Future<T> submit(Object arg, Runnable task, T result) {return choose(arg).submit(task, result);}
    public <T> Future<T> submit(Object arg, Callable<T> task) {return choose(arg).submit(task);}

    /** 按照一定规则来选择一个Dtp执行器 */
    protected DtpExecutor choose(Object arg) {
        int size = this.executors.size();
        if (size == 1) return this.executors.get(0); // 如果只有一个Dtp执行器，则将其直接返回即可

        // 根据arg的哈希值取余来选择某个Dtp执行器，如果arg为空则默认取count+1的值，然后通过取余来选择某个Dtp执行器
        int index;
        if (Objects.isNull(arg)) {
            int i = count.getAndIncrement(); // count值+1
            if (i < 0) { // 越界
                i = 0;
                count.set(0);
            }
            index = i;
        } else {
            index = arg.hashCode(); // 哈希值
        }
        return this.executors.get(index % size);
    }

    /** 设置核心线程数量，不能小于设置前的旧值 */
    @Override
    public void setCorePoolSize(int corePoolSize) {
        if (corePoolSize < this.executors.size()) {
            log.error("Except corePoolSize must >= {}, newCorePoolSize: {}, threadPoolName: {}", this.executors.size(), corePoolSize, threadPoolName);
            return;
        }

        // 如果设置的corePoolSize数量大于原本executors中执行器的数量，则创建新的Dtp执行器补齐缺少的数量
        for (int i = this.executors.size(); i < corePoolSize; i++) {
            this.executors.add(new DtpExecutor(1, 1, getKeepAliveTime(TimeUnit.SECONDS), TimeUnit.SECONDS,
                    buildLbq(getQueueName(), getQueueCapacity()), buildThreadFactory(i),
                    RejectHandlerGetter.getProxy(getRejectHandlerName()))); // 获得代理对象
        }
    }

    /** 获得dtp执行器总数量 */
    @Override
    public int getCorePoolSize() {return this.executors.size();}
    @Override
    public final int getPoolSize() {return getCorePoolSize();}
    @Override
    public final int getMaximumPoolSize() {return getCorePoolSize();}

    /** 关闭所有dtp执行器，实际上调用ThreadPoolExecutor#shutdown() */
    @Override
    public void shutdown() {for (ExecutorService executor : this.executors) executor.shutdown();}

    /** 立刻关闭所有dtp执行器，调用ThreadPoolExecutor#shutdownNow() */
    @Override
    public List<Runnable> shutdownNow() {
        return this.executors.stream().map(ExecutorService::shutdownNow).flatMap(Collection::stream).collect(Collectors.toList());
    }

    /** 判断所有的Dtp执行器是否都已经关闭了，实际上调用ThreadPoolExecutor#isShutdown() */
    @Override
    public boolean isShutdown() {
        boolean result = true;
        for (ExecutorService executor : this.executors)
            result = result && executor.isShutdown();
        return result;
    }

    /** 判断所有的Dtp执行器是否都已经被终止了，实际上调用ThreadPoolExecutor#isTerminated() */
    @Override
    public boolean isTerminated() {
        boolean result = true;
        for (ExecutorService executor : this.executors)
            result = result && executor.isTerminated();
        return result;
    }

    /** 判断所有的Dtp执行器在timeout时间内，是否都已经被终止了，实际上调用ThreadPoolExecutor#awaitTermination(timeout, unit) */
    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        boolean result = true;
        for (ExecutorService executor : this.executors)
            result = result && executor.awaitTermination(timeout, unit);
        return result;
    }

    /** 获得ThreadFactory，实际上调用ThreadPoolExecutor#getThreadFactory() */
    private ThreadFactory buildThreadFactory(int index) {
        if (getThreadFactory() instanceof NamedThreadFactory) {
            String prefix = ((NamedThreadFactory) getThreadFactory()).getNamePrefix() + "#" + index;
            return new NamedThreadFactory(prefix); // 如果是NamedThreadFactory，需要额外处理一下namePrefix
        }
        return getThreadFactory();
    }
}
